package com.duobaoyu.dby.supply.biz.service.queue.service;

import com.alibaba.fastjson.JSON;
import com.duobaoyu.dby.channel.adapter.enums.DistributionQueueJobStatusEnum;
import com.duobaoyu.dby.supply.biz.service.queue.core.DistributionQueueMessage;
import com.duobaoyu.dby.supply.core.delay.config.RedissonTemplate;
import com.duobaoyu.dby.supply.db.entity.message.DelayMessage;
import com.duobaoyu.dby.supply.db.repository.message.DelayMessageRepository;
import com.google.common.collect.Maps;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.codec.digest.DigestUtils;
import org.springframework.dao.DuplicateKeyException;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;

import java.util.HashMap;

import static com.duobaoyu.dby.supply.biz.service.queue.core.DistributionQueueConstant.DISTRIBUTION_QUEUE_MESSAGE_ID;

/**
 * Created by shuming on 2023/9/27 14:32
 * 任务service
 */
@Slf4j
@Component
@RequiredArgsConstructor
public class DistributionQueueJobService {

    private final RedissonTemplate redissonTemplate;
    private final DelayMessageRepository delayMessageRepository;


    @Transactional(propagation = Propagation.REQUIRES_NEW, rollbackFor = Exception.class)
    public void send(DistributionQueueMessage distributionQueueMessage) {
        this.send(distributionQueueMessage, Boolean.TRUE);
    }

    public void send(DistributionQueueMessage distributionQueueMessage, boolean isMessageStore) {
        if(log.isDebugEnabled()) {
            log.debug("Distribution Queue Message Send : {}, isMessageStore : {}", JSON.toJSONString(distributionQueueMessage), isMessageStore);
        }
        String msgId = DigestUtils.md5Hex(distributionQueueMessage.toString());
        if (isMessageStore) {
            this.store(distributionQueueMessage, msgId);
        }
        HashMap<String, Object> headers = Maps.newHashMap();
        headers.put(DISTRIBUTION_QUEUE_MESSAGE_ID, msgId);
        if (distributionQueueMessage.getTriggerTime() > System.currentTimeMillis()) {
            redissonTemplate.sendWithDelay(distributionQueueMessage.getTopicName(),
                    distributionQueueMessage, headers, distributionQueueMessage.getDelayTime());
        } else {
            redissonTemplate.send(distributionQueueMessage.getTopicName(), distributionQueueMessage, headers);
        }
    }

    private void store(DistributionQueueMessage distributionQueueMessage, String msgId) {
        DelayMessage delayMessage = new DelayMessage();
        delayMessage.setUniqueKey(msgId);
        delayMessage.setMessageType(distributionQueueMessage.getBizScenario());
        delayMessage.setMessage(JSON.toJSONString(distributionQueueMessage));
        delayMessage.setDelayTime(distributionQueueMessage.getTriggerTime());
        delayMessage.setStatus(DistributionQueueJobStatusEnum.WAIT_HANDLE.getValue());
        delayMessage.setOperationalCode(distributionQueueMessage.getBizCode());
        try {
            delayMessageRepository.save(delayMessage);
        } catch (DuplicateKeyException ignored) {

        }
    }
}
